package com.niit.sparkSql


import java.util.Properties

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author gaomj
 * @data 2022/12/14 23:13
 */
object SparkSQL  {

  // 使用样例类 ，getter 和 setter  ，去定义 person
  case class StationLog(
                         ID: String, //ID
                         Name:String,//name
                         Sex: Int, //sex
                         Date: Int, //date
                         num: String, //学号
                         Status: String, //状态
                         Term: Int //学期
                       )

  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "D:\\code\\hadoop-2.7.3")
    System.setProperty("HADOOP_USER_NAME", "root")


    // spark core ： sparkconf  加载配置 是否本地运行，加载应用程序名称 || spark context   sc
    // spark sql   : sqlContext  sql 提供一个编译的环境， 从依赖包 spark-sql

    // 光标前后切换  home键，end键
    val conf = new SparkConf().setMaster("local[*]").setAppName("helloSparkSql")
    val sc = new SparkContext(conf)

    // spark sql
    val sqlcon = new SQLContext(sc)


    //txt文件
    val studentRDD = sc.textFile("D:\\student.txt").map(x => {
      // 原始数据：  logdatashow产生的
      // split “ ，” 切分
      val line = x.split(",")
      (line(0).toString, line(1).toString, line(2).toInt, line(3).toInt, line(4).toString, line(5).toString, line(6).toInt)
    })
    // studentRDD   元祖
    // studentRDD      person 映射
    // x 是  studentRDD 当中每个元素 ，每个元素是  元祖   ( 205818,Aesrgf,1,20221219,1000000001,L,6 ) ，第一个值是 下滑线 ._1
    val personRdd = studentRDD.map(x => StationLog(x._1, x._2, x._3,x._4,x._5,x._6,x._7))

    //数据在控制台展示
    import sqlcon.implicits._
    personRdd.toDF.show()


    //spark sql语句
    //ID Sex count
    val personDF = personRdd.toDF
    personDF.registerTempTable("person")
    val personSql = sqlcon.sql("select ID,Sex,count(*) as count from person group by ID,Sex ")
    personSql.show()

    //连接MySQL数据库
    val prop = new Properties()
    prop.put("user", "root")
    prop.put("password", "mmm18623908829")


    //将数据追加到数据库
    //overwrite可以实现数据更新覆盖
    personSql.write.mode("overwrite").jdbc("jdbc:mysql://localhost:3306/echarts", "echarts.echarts", prop)

    //停止SparkContext
    sc.stop()

  }
}
